package com.zjh.reactive;

import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;

/**
 * @Auther: zjh
 * @Date: 2025/11/3 - 11 - 03 - 12:59
 * @Description: com.zjh.reactive
 * @version: 1.0
 */

public class FilterDemo {
    @Test
    public void filter(){
        Flux.range(1,5)
                .filter(e->e%2==0)
                .subscribe(System.out::println);
    }
    @Test
    public void take(){
        Flux.range(1,5)
                .take(3)
                .subscribe(System.out::println);
    }
    @Test
    public void takeLast(){
        Flux.range(1,5)
                //5秒之内处理完
                .take(Duration.ofSeconds(5))
                .subscribe(System.out::println);
        Flux.range(1,5)
                .takeLast(3)
                .subscribe(System.out::println);
    }
    @Test
    public void elementAt(){
        Flux.range(1,3)
                .elementAt(2)
                .subscribe(System.out::println);
    }
    @Test
    public void takeAndSkip() throws InterruptedException {
        Flux.interval(Duration.ofSeconds(1))
                .skipUntilOther(Mono.just("start").delayElement(Duration.ofSeconds(1)))
                .takeUntilOther(Mono.just("stop").delayElement(Duration.ofSeconds(5)))
                .subscribe(System.out::println);
        Thread.sleep(6000);
    }
}

